package edu.purdue.cs.aggr.messageprocessor;

import edu.purdue.cs.aggr.AggrException;
import edu.purdue.cs.aggr.Main;
import edu.purdue.cs.aggr.Vertex;
import edu.purdue.cs.aggr.VertexManager;
import edu.purdue.cs.aggr.message.ConnectionEstablishmentMessage;
import edu.purdue.cs.aggr.message.Message;
import edu.purdue.cs.aggr.message.MessageSender;
import edu.purdue.cs.aggr.message.VertexSetupMessage;

public class VertexSetupMessageProcessor extends MessageProcessor {

  private VertexManager manager = null;
	
	public VertexSetupMessageProcessor (VertexManager m) {
		this.manager = m;
	}
	
	@Override
	public void processMessage(Message message) throws AggrException {
		System.out.println("Processing vertex setup message...");
		VertexSetupMessage setUpMessage = (VertexSetupMessage) message;
		System.out.println("Source vertices:" + setUpMessage.getSourceVertices().size());
		Vertex vertex = manager.getVertex(message.getDestination()); 
		vertex.setDestinationVertex(setUpMessage.getDestinationVertex());
		if (setUpMessage.getDestinationVertex() != null) {
		    establishConnectionWithNode(setUpMessage.getDestinationVertex());
		}
		System.out.println("Destination set to :" + vertex.getDestinationVertex());
		System.out.println("Leaf :" + setUpMessage.isInputReader());
		vertex.setSourceNodes(setUpMessage.getSourceVertices());
		
		if (setUpMessage.isInputReader()) {
			vertex.setHdfsNamenode(setUpMessage.getHdfsNamenode());
			vertex.setHdfsFilename(setUpMessage.getInputFile());
			vertex.setLeafVertex(true);
			vertex.setAccumulatorType(setUpMessage.getAccumulatorType());
		}

		vertex.setAccumulatorSubtype(setUpMessage.getAccumulatorSubtype());
		
		if (vertex.isLeafVertex() || (setUpMessage.getDestinationVertex() == null)) { // Leaf or root
			System.out.println("Establishing a connection with controller...");
			establishConnectionWithNode(vertex.getControllerId());
		} else {
			System.out.println("Did NOT establish a connection with controller...");
		}
		
		vertex.setSetup(true);
		if (setUpMessage.isInputReader()) {
			synchronized (vertex) {
				vertex.notifyAll();
			}
		}
	}

	private void establishConnectionWithNode(String nodeId) {
	  Message m = new ConnectionEstablishmentMessage();
	  m.setDestination(nodeId);
		manager.sendMessage(m);
	}
}
